package com.ctw.mapmycity.hbase.controller;

import java.io.IOException;
import java.util.List;

import javax.annotation.Resource;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.hadoop.hbase.HbaseTemplate;
import org.springframework.stereotype.Component;

import com.ctw.mapmycity.hbase.repository.PostRepository;
import com.ctw.mapmycity.model.PostModel;

@Component
public class PostController implements InitializingBean {
	
	private String tableName = "post";
	private byte[] tableNameAsBytes = Bytes.toBytes(tableName);

	@Resource(name = "hbaseConfiguration")
	private Configuration config;

	@Autowired
	private HbaseTemplate hbaseTemplate;

	@Autowired
	private PostRepository postRepository;

	private HBaseAdmin admin;

	public void initialize() throws IOException {

		if (admin.tableExists(tableNameAsBytes)) {
//			return;
			if (!admin.isTableDisabled(tableNameAsBytes)) {
				System.out.printf("Disabling %s\n", tableName);
				admin.disableTable(tableNameAsBytes);
			}
			System.out.printf("Deleting %s\n", tableName);
			admin.deleteTable(tableNameAsBytes);
		}

		HTableDescriptor tableDescriptor = new HTableDescriptor(tableName);
		tableDescriptor.addFamily(new HColumnDescriptor(PostRepository.CF_INFO));
		tableDescriptor.addFamily(new HColumnDescriptor(PostRepository.CF_AGREE));
		tableDescriptor.addFamily(new HColumnDescriptor(PostRepository.CF_SHARE));

		admin.createTable(tableDescriptor);

	}

	public void save(PostModel model) {
		postRepository.save(model);
	}
	
	public PostModel get(String ts) {
		return postRepository.get(ts);
	}

	public List<PostModel> findAll() {
		return postRepository.findAll();
	}
	
	public List<PostModel> find(String[] opts) {
		return postRepository.find(opts);
	}
	
	public void delete(String ts) {
		postRepository.delete(ts);
	}
	
	public String confirm (String ts, String email) {
		return postRepository.confirm(ts, email);
	}
	
	public List<String> getConfirmList (String ts) {
		return postRepository.getConfirmationList(ts);
	}
	
	@Override
	public void afterPropertiesSet() throws Exception {
		admin = new HBaseAdmin(config);
	}
}